package com.bw.gmall.realtime.dws.app;

import cn.hutool.core.collection.ListUtil;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.UserLoginBean;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.DateFormatUtil;
import com.bw.gmall.realtime.common.util.DorisMapFunction;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import javafx.scene.input.DataFormat;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.jar.JarEntry;

/**
 * @projectName: gmalll2206
 * @package: com.bw.gmall.realtime.dws.app
 * @className: DwsUserUserLoginWindow
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/3 13:47
 * @version: 1.0
 */

public class DwsUserUserLoginWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsUserUserLoginWindow().start(Constant.TOPIC_DWD_TRAFFIC_PAGE, Constant.DWS_USER_USER_LOGIN_WINDOW, 4, 10024);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        //保留 uid 不为 null 且 last_page_id 为 null 或 last_page_id 为 login 的浏览记录。
        SingleOutputStreamOperator<JSONObject> getEtl = getGetEtl(dataStreamSource);
//        getEtl.print();
        //设置水位线
        SingleOutputStreamOperator<JSONObject> wmStream = getTs(getEtl);
        //按照uid分组
        KeyedStream<JSONObject, String> keyByStream = getKeyByStream(wmStream);
        //6）统计回流用户数和独立用户数
        SingleOutputStreamOperator<UserLoginBean> process = getProcess(keyByStream);
        //聚合
        SingleOutputStreamOperator<UserLoginBean> process1 = getProcess(process);

//        process1.print();
//        process1.map(new DorisMapFunction<>()).print();
        process1.map(new DorisMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink(Constant.DWS_USER_USER_LOGIN_WINDOW));
    }

    private static SingleOutputStreamOperator<UserLoginBean> getProcess(SingleOutputStreamOperator<UserLoginBean> process) {
        return process.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessAllWindowFunction<UserLoginBean, UserLoginBean, TimeWindow>() {
                    @Override
                    public void process(ProcessAllWindowFunction<UserLoginBean, UserLoginBean, TimeWindow>.Context context, Iterable<UserLoginBean> iterable, Collector<UserLoginBean> collector) throws Exception {
                        ArrayList<UserLoginBean> list = ListUtil.toList(iterable);
                        long uuCt = list.stream().mapToLong(e -> e.getUuCt()).sum();
                        long backCt = list.stream().mapToLong(e -> e.getBackCt()).sum();
                        long start = context.window().getStart();
                        String start_time = DateFormatUtil.tsToDateTime(start);
                        long end = context.window().getEnd();
                        String end_time = DateFormatUtil.tsToDateTime(end);
                        UserLoginBean next = list.iterator().next();
                        String s = DateFormatUtil.tsToDate(System.currentTimeMillis());
                        collector.collect(UserLoginBean.builder().stt(start_time).edt(end_time).curDate(s)
                                .backCt(backCt).uuCt(uuCt).ts(next.getTs()).
                                build());
                    }
                });
    }

    private static SingleOutputStreamOperator<UserLoginBean> getProcess(KeyedStream<JSONObject, String> keyByStream) {
        return keyByStream.process(new KeyedProcessFunction<String, JSONObject, UserLoginBean>() {
            private ValueState<String> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> state = new ValueStateDescriptor<>("valueState", String.class);
                valueState = getRuntimeContext().getState(state);
            }

            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, UserLoginBean>.Context context, Collector<UserLoginBean> collector) throws Exception {
                Long ts = jsonObject.getLong("ts");
                String s = DateFormatUtil.tsToDate(ts);
                String value = valueState.value();
                // 回流
                long backCt = 0L;
                // 独立登录用户数
                long uuCt = 0L;
                if (value != null) {
                    if (!s.equals(value)) {
                        uuCt = 1L;
                        valueState.update(s);
                    }
                    if (ts - DateFormatUtil.dateToTs(value) > 7 * 24 * 60 * 60 * 1000) {
                        backCt = 1L;
                    }
                } else if (value == null) {
                    uuCt = 1L;
                    backCt = 0L;
                    valueState.update(s);
                }

                if (uuCt != 0 || backCt != 0) {
                    UserLoginBean build = UserLoginBean.builder().uuCt(uuCt).backCt(backCt).ts(ts).build();
                    collector.collect(build);
                }

            }
        });
    }

    private static KeyedStream<JSONObject, String> getKeyByStream(SingleOutputStreamOperator<JSONObject> wmStream) {
        return wmStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getJSONObject("common").getString("uid");
            }
        });
    }

    private static SingleOutputStreamOperator<JSONObject> getTs(SingleOutputStreamOperator<JSONObject> getEtl) {
        return getEtl.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject jsonObject, long l) {
                return jsonObject.getLong("ts");
            }
        }));
    }

    private static SingleOutputStreamOperator<JSONObject> getGetEtl(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.flatMap(new RichFlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                if (s != null) {
                    JSONObject jsonObject = JSONObject.parseObject(s);
                    JSONObject common = jsonObject.getJSONObject("common");
                    JSONObject page = jsonObject.getJSONObject("page");
                    String ts = jsonObject.getString("ts");
                    if (common != null && page != null && ts != null) {
                        String uid = common.getString("uid");
                        String last_page_id = page.getString("last_page_id");
                        if (uid != null && (last_page_id != null || "login".equals(page))) {
                            collector.collect(jsonObject);
                        }
                    }
                }
            }
        });
    }
}
